package com.lagou.hbase.client;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

/*
 * @className MyProcessor1
 * @description
 * @author admin
 * @date 2022/7/9 15:22
 * */
public class MyProcessor1 extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        HTableInterface friend = e.getEnvironment().getTable(TableName.valueOf("friend"));
        // 获取rowKey
        byte[] rowKeyUid = delete.getRow();
        // 获取所有cell对象
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        Set<Map.Entry<byte[], List<Cell>>> entrySet = familyCellMap.entrySet();
        for (Map.Entry<byte[], List<Cell>> entry : entrySet) {
            // 获取列族信息
            System.out.println(new String(entry.getKey()));
            // 改列族所有单元格
            List<Cell> cells = entry.getValue();
            // uid1  列族friends 列信息uid2 数据uid2
            // uid2  列族friends 列信息uid1 数据uid1
            // uid1 删除uid2
            // uid2 删除uid1
            for (Cell cell : cells) {
                // 获得rowKey->uid
                byte[] rowKey = CellUtil.cloneRow(cell);
                // 列信息
                byte[] column = CellUtil.cloneQualifier(cell);
                // 判断要删除的目标数据是否存在，存在则删除
                boolean flag = friend.exists(new Get(column).addColumn(Bytes.toBytes("friends"), rowKey));
                if(flag){
                    Delete myDelete = new Delete(column).addColumn(Bytes.toBytes("friends"), rowKey);
                    // 执行删除
                    friend.delete(myDelete);
                }
            }
        }

    }

}
